package com.zhang.hadoop.testMain;

import com.alibaba.fastjson.JSONObject;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import static java.util.stream.Collectors.toList;

/**
 * @author zhang yufei
 * @date 2022/4/26 3:06 下午
 **/
public class Test7 {

    public static List<Integer> numList = Arrays.asList(
            10077000,
            7206000,
            3041000,
            3005000,
            2758000,
            2567000,
            2188000,
            2081000,
            1653000,
            1652000,
            1646000,
            1497000,
            945000,
            813000,
            796000,
            599000,
            587000,
            561000,
            561000,
            545000,
            436000,
            351000,
            281000,
            219000,
            155000,
            116000,
            100000,
            88000,
            80000,
            77000,
            65000,
            65000,
            63000,
            60000,
            53000,
            49000,
            41000,
            40000,
            39000,
            39000,
            36000,
            34000,
            24000,
            23000,
            22000,
            21000,
            21000,
            19000,
            16000,
            13000,
            12000,
            11000,
            11000,
            10000,
            9908,
            9513,
            8927,
            7763,
            7699,
            6824,
            6296,
            6207,
            5692,
            5612,
            3535,
            3078,
            2954,
            2933,
            2902,
            2732,
            2583,
            2544,
            2475,
            2150,
            2140,
            1969,
            1897,
            1757,
            1725,
            1492,
            1289,
            1285,
            1233,
            1006,
            617,
            546,
            499,
            450,
            354,
            353,
            304,
            302,
            259,
            254,
            205,
            186,
            170,
            163,
            161,
            148,
            126,
            72,
            60,
            55,
            46,
            46,
            46,
            41,
            41,
            39,
            34,
            32,
            32,
            32,
            30,
            30,
            22,
            22,
            20,
            16,
            15,
            15,
            14,
            13,
            11,
            9,
            8,
            6,
            4,
            4,
            2,
            2,
            2,
            1,
            1,
            1,
            1,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0,
            0
    );

    public static List<String> nodeList = Collections.synchronizedList(new ArrayList<>());

    public static List<String> nodeListCopy = Collections.synchronizedList(new ArrayList<>());

    public static Map<String, Integer> topicMap = Collections.synchronizedMap(new LinkedHashMap<String, Integer>());

    public static Map<String, Integer> slotMap = Collections.synchronizedMap(new LinkedHashMap<String, Integer>());

    public static Map<String, Integer> slotMessageNumMap = Collections.synchronizedMap(new LinkedHashMap<String, Integer>());

    public static Map<String, List<Map<String, String>>> relationMap = Collections.synchronizedMap(new LinkedHashMap<String, List<Map<String, String>>>());

    public static Random random = new Random();

    public static String NODE = "node-";

    public static volatile Integer number = 0;

    public static volatile Integer change = 0;

    public static String TOPIC = "topic-";

    public static String SLOT = "slot-";

    public static Integer SIZE = 5 * 10000;

    public static void main(String[] args) throws Exception {
        for (int i = 1; i <= 10; i++) {
            nodeList.add(NODE + i);
            number = number + 1;
        }
        nodeListCopy.addAll(nodeList);
        for (int i = 1; i <= 200; i++) {
            topicMap.put(TOPIC + i, numList.get(i - 1));
        }
        for (Map.Entry<String, Integer> entry : topicMap.entrySet()) {
            Integer num = entry.getValue();
            Integer slot = num / SIZE;
            for (int i = 1; i <= slot; i++) {
                slotMessageNumMap.put(entry.getKey() + "-" + SLOT + i, SIZE);
            }
            if (num % SIZE > 0) {
                slot = slot + 1;
                slotMessageNumMap.put(entry.getKey() + "-" + SLOT + slot, num % SIZE);
            }
            if (slot < 2) {
                slot = 2;
                if (num == 0) {
                    slotMessageNumMap.put(entry.getKey() + "-" + SLOT + 1, 0);
                    slotMessageNumMap.put(entry.getKey() + "-" + SLOT + 2, 0);
                } else {
                    slotMessageNumMap.put(entry.getKey() + "-" + SLOT + 1, num / 2);
                    slotMessageNumMap.put(entry.getKey() + "-" + SLOT + 2, num / 2);
                }
            }
            slotMap.put(entry.getKey(), slot);
            for (int i = 1; i <= slot; i++) {
                String slotName = entry.getKey() + "-" + SLOT + i;
                int i1 = getPosition(nodeList.size());
                String node = nodeList.get(i1);
                if (relationMap.containsKey(node)) {
                    List<Map<String, String>> relation = relationMap.get(node);
                    Map<String, String> map = new HashMap<>();
                    map.put(entry.getKey(), slotName);
                    relation.add(map);
                } else {
                    List<Map<String, String>> relation = new ArrayList<>();
                    Map<String, String> map = new HashMap<>();
                    map.put(entry.getKey(), slotName);
                    relationMap.put(node, relation);
                }
            }
        }
        System.out.println(JSONObject.toJSONString(nodeList));
        System.out.println(JSONObject.toJSONString(topicMap));
        System.out.println(JSONObject.toJSONString(slotMap));
        //System.out.println(JSONObject.toJSONString(relationMap));
        System.out.println("**** relationMap 初始化 ****");
        for (Map.Entry<String, List<Map<String, String>>> entry : relationMap.entrySet()) {
            Integer slot = 0;
            slot = slot + entry.getValue().size();
            Integer messageNum = 0;
            for (Map<String, String> topicAndSlot : entry.getValue()) {
                for (String topic : topicAndSlot.keySet()) {
                    messageNum = messageNum + slotMessageNumMap.get(topicAndSlot.get(topic));
                }
            }
            System.out.println(entry.getKey() + "\t" + slot + "\t" + messageNum + "\t" + JSONObject.toJSONString(entry.getValue()));
        }
        System.out.println("**** relationMap 初始化 ****");

        Thread threadChange = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(5 * 1000);
                        boolean flag = random.nextBoolean();
                        //boolean flag = true;
                        //boolean flag = false;
                        int num = getPosition(3) + 1;
                        for (int i = 0; i < num; i++) {
                            if (flag) {
                                //新增节点
                                number = number + 1;
                                nodeList.add(NODE + number);
                            } else {
                                //宕机节点
                                if (nodeList.size() == 1) {
                                    System.out.println("节点不足，停止缩容");
                                    continue;
                                }
                                int index = getPosition(nodeList.size());
                                nodeList.remove(index);
                            }
                        }
                        if (flag) {
                            System.out.println("++++扩容" + num + "台++++");
                        } else {
                            System.out.println("----缩容" + num + "台----");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        change = 1;
                    }
                }
            }
        });
        threadChange.start();

        Thread threadListener = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    if (change == 1) {
                        try {
                            List<String> reduceAdd = nodeList.stream().filter(item -> !nodeListCopy.contains(item)).collect(toList());
                            Float oldNum = ((Integer) nodeListCopy.size()).floatValue();
                            Float newNum = ((Integer) nodeList.size()).floatValue();
                            Float f1 = newNum / oldNum;
                            if (!reduceAdd.isEmpty()) {
                                for (String node : nodeListCopy) {
                                    List<Map<String, String>> topicList = relationMap.get(node);
                                    Float f2 = ((Integer) topicList.size()).floatValue();
                                    Float f3 = f2 / f1;
                                    Float f4 = f2 - f3;
                                    Integer num = f4.intValue();
                                    for (int i = 0; i < num; i++) {
                                        int index = getPosition(topicList.size());
                                        Map<String, String> map = new HashMap<>();
                                        for (Map.Entry<String, String> entry : topicList.get(index).entrySet()) {
                                            map.put(entry.getKey(), entry.getValue());
                                        }
                                        topicList.remove(index);
                                        int nodeIndex = getPosition(reduceAdd.size());
                                        String nodeName = reduceAdd.get(nodeIndex);
                                        if (relationMap.containsKey(nodeName)) {
                                            relationMap.get(nodeName).add(map);
                                        } else {
                                            List<Map<String, String>> list = new ArrayList<>();
                                            list.add(map);
                                            relationMap.put(nodeName, list);
                                        }
                                    }
                                }
                                nodeListCopy.clear();
                                nodeListCopy.addAll(nodeList);
                            }
                            List<String> reduceDelete = nodeListCopy.stream().filter(item -> !nodeList.contains(item)).collect(toList());
                            if (!reduceDelete.isEmpty()) {
                                List<String> reduceAlive = nodeList.stream().filter(item -> nodeListCopy.contains(item)).collect(toList());
                                for (String node : reduceDelete) {
                                    List<Map<String, String>> topicList = relationMap.get(node);
                                    for (Map<String, String> map : topicList) {
                                        Map<String, String> topicSlot = new HashMap<>();
                                        for (Map.Entry<String, String> entry : map.entrySet()) {
                                            topicSlot.put(entry.getKey(), entry.getValue());
                                        }
                                        int index = getPosition(reduceAlive.size());
                                        relationMap.get(reduceAlive.get(index)).add(topicSlot);
                                    }
                                    relationMap.remove(node);
                                }
                                nodeListCopy.clear();
                                nodeListCopy.addAll(nodeList);
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            change = 2;
                        }
                    }
                }
            }
        });
        threadListener.start();

        while (true) {
            if (change == 2) {
                //System.out.println("**** relationMap begin ****");
                //for (Map.Entry<String, List<Map<String, String>>> entry : relationMap.entrySet()) {
                //    Integer slot = 0;
                //    slot = slot + entry.getValue().size();
                //    Integer messageNum = 0;
                //    for (Map<String, String> topicAndSlot : entry.getValue()) {
                //        for (String topic : topicAndSlot.keySet()) {
                //            messageNum = messageNum + slotMessageNumMap.get(topicAndSlot.get(topic));
                //        }
                //    }
                //    System.out.println(entry.getKey() + "\t" + slot + "\t" + messageNum + "\t" + JSONObject.toJSONString(entry.getValue()));
                //}
                //System.out.println("**** relationMap end ****");
                System.out.println("**** topicMap begin ****");
                for (String topicName : topicMap.keySet()) {
                    List<Map<String, Integer>> nodeList = new ArrayList<>();
                    for (Map.Entry<String, List<Map<String, String>>> entry : relationMap.entrySet()) {
                        for (Map<String, String> map : entry.getValue()) {
                            if (map.containsKey(topicName)) {
                                boolean flag = true;
                                for (Map<String, Integer> node : nodeList) {
                                    if (node.containsKey(entry.getKey())) {
                                        node.put(entry.getKey(), node.get(entry.getKey()) + slotMessageNumMap.get(map.get(topicName)));
                                        flag = false;
                                    }
                                }
                                if (flag) {
                                    Map<String, Integer> node = new HashMap<>();
                                    node.put(entry.getKey(), slotMessageNumMap.get(map.get(topicName)));
                                    nodeList.add(node);
                                }
                            }
                        }
                    }
                    System.out.println(topicName + "\t" + JSONObject.toJSONString(nodeList));
                }
                System.out.println("**** topicMap end ****");
                change = 0;
            }
        }
    }

    public static Integer getPosition(Integer length) {
        String type = "random";
        if (type.equals("random")) {
            Integer result = random.nextInt(length);
            return result;
        }
        return null;
    }
}
